Skip to content

Add lambda support and array_transform udf#21679

Merged
rluvaton merged 88 commits intoapache:mainfrom
gstvg:lambda_and_array_transform
Apr 28, 2026
Merged

Add lambda support and array_transform udf#21679
rluvaton merged 88 commits intoapache:mainfrom
gstvg:lambda_and_array_transform

Conversation

@gstvg
Copy link
Copy Markdown
Contributor

@gstvg gstvg commented Apr 16, 2026

This a clean version of #18921 to make it easier to review

this is a breaking change due to adding variant to Expr enum, new methods on traits Session, FunctionRegistry and ContextProvider and a new arg on TaskContext::new

This PR adds support for lambdas and the array_transform function used to test the lambda implementation.

Example usage:

SELECT array_transform([2, 3], v -> v != 2);

[false, true]

-- arbitrally nested lambdas are also supported
SELECT array_transform([[[2, 3]]], m -> array_transform(m, l -> array_transform(l, v -> v*2)));

[[[4, 6]]]

Note: column capture has been removed for now and will be added on a follow on PR, see #21172

Some comments on code snippets of this doc show what value each struct, variant or field would hold after planning the first example above. Some literals are simplified pseudo code

3 new Expr variants are added, HigherOrderFunction, owing a new trait HigherOrderUDF, which is like a ScalarFunction/ScalarUDFImpl with support for lambdas, Lambda, for the lambda body and it's parameters names, and LambdaVariable, which is like Column but for lambdas parameters.

Their logical representations:

enum Expr {
    // array_transform([2, 3], v -> v != 2)
    HigherOrderFunction(HigherOrderFunction),
    // v -> v != 2
    Lambda(Lambda),
    // v, of the lambda body: v != 2
    LambdaVariable(LambdaVariable),
   ...
}

// array_transform([2, 3], v -> v != 2)
struct HigherOrderFunction {
    // global instance of array_transform
    pub func: Arc<dyn HigherOrderUDF>,
    // [Expr::ScalarValue([2, 3]), Expr::Lambda(v -> v != 2)]
    pub args: Vec<Expr>,
}

// v -> v != 2
struct Lambda {
    // ["v"]
    pub params: Vec<String>,
    // v != 2
    pub body: Box<Expr>,
}

// v, of the lambda body: v != 2
struct LambdaVariable {
    // "v"
    pub name: String,
    // Field::new("", DataType::Int32, false) 
    // Note: a follow on PR will make this field optional
    // to free expr_api from specifying it beforehand, 
    // and add resolve_lambda_variables method to Expr,
    // similar to Expr::Placeholder, see #21172
    pub field: FieldRef, 
    pub spans: Spans,
}

The example would be planned into a tree like this:

HigherOrderFunctionExpression
  name: array_transform
  children:
    1. ListExpression [2,3]
    2. LambdaExpression
         parameters: ["v"]
         body:
            BinaryExpression (!=)
              left:
                 LambdaVariableExpression("v", Field::new("", Int32, false))
              right:
                 LiteralExpression("2")

The physical counterparts definition:

struct HigherOrderFunctionExpr {
    // global instance of array_transform
    fun: Arc<dyn HigherOrderUDF>,
    // "array_transform"
    name: String,
    // [LiteralExpr([2, 3], LambdaExpr("v -> v != 2"))]
    args: Vec<Arc<dyn PhysicalExpr>>,
    // [1], the positions at args that contains lambdas
    lambda_positions: Vec<usize>,
    // Field::new("", DataType::new_list(DataType::Boolean, false), false)
    return_field: FieldRef,
    config_options: Arc<ConfigOptions>, 
}


struct LambdaExpr {
    // ["v"]
    params: Vec<String>,
    // v -> v != 2
    body: Arc<dyn PhysicalExpr>,
}

struct LambdaVariable {
    // Field::new("v", DataType::Int32, false)
    field: FieldRef,
    // 0, the first and only parameter, "v"
    index: usize,
}

Note: For those who primarly wants to check if this lambda implementation supports their usecase and don't want to spend much time here, it's okay to skip most collapsed blocks, as those serve mostly to help code reviewers, with the exception of HigherOrderUDF and the array_transform implementation of HigherOrderUDF relevant methods, collapsed due to their size

The added HigherOrderUDF trait is almost a clone of ScalarUDFImpl, with the exception of:

  1. return_field_from_args and invoke_with_args, where now args.args is a list of enums with two variants: Value or Lambda instead of a list of values
  2. the addition of lambda_parameters, which return a Field for each parameter supported for every lambda argument based on the Field of the non lambda arguments
  3. the removal of return_field and the deprecated ones is_nullable and display_name.
  4. Not yet includes analogues to the methods preimage, placement, evaluate_bounds, propagate_constraints, output_ordering and preserves_lex_ordering
HigherOrderUDF
trait HigherOrderUDF {
    /// Return the field of all the parameters supported by all the supported lambdas of this function
    /// based on the field of the value arguments. If a lambda support multiple parameters, or if multiple
    /// lambdas are supported and some are optional, all should be returned,
    /// regardless of whether they are used on a particular invocation
    ///
    /// Tip: If you have a [`HigherOrderFunction`] invocation, you can call the helper
    /// [`HigherOrderFunction::lambda_parameters`] instead of this method directly
    ///
    /// [`HigherOrderFunction`]: crate::expr::HigherOrderFunction
    /// [`HigherOrderFunction::lambda_parameters`]: crate::expr::HigherOrderFunction::lambda_parameters
    ///
    /// Example for array_transform:
    ///
    /// `array_transform([2.0, 8.0], v -> v > 4.0)`
    ///
    /// ```ignore
    /// let lambda_parameters = array_transform.lambda_parameters(&[
    ///      Arc::new(Field::new("", DataType::new_list(DataType::Float32, false))), // the Field of the literal `[2, 8]`
    /// ])?;
    ///
    /// assert_eq!(
    ///      lambda_parameters,
    ///      vec![
    ///         // the lambda supported parameters, regardless of how many are actually used
    ///         vec![
    ///             // the value being transformed
    ///             Field::new("", DataType::Float32, false),
    ///         ]
    ///      ]
    /// )
    /// ```
    ///
    /// The implementation can assume that some other part of the code has coerced
    /// the actual argument types to match [`Self::signature`].
    fn lambda_parameters(&self, value_fields: &[FieldRef]) -> Result<Vec<Vec<Field>>>;
    fn return_field_from_args(&self, args: LambdaReturnFieldArgs) -> Result<FieldRef>;
    fn invoke_with_args(&self, args: HigherOrderFunctionArgs) -> Result<ColumnarValue>;
   // ... omitted methods that are similar in ScalarUDFImpl
}

/// An argument to a lambda function
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ValueOrLambda<V, L> {
    /// A value with associated data
    Value(V),
    /// A lambda with associated data
    Lambda(L),
}

/// Information about arguments passed to the function
///
/// This structure contains metadata about how the function was called
/// such as the type of the arguments, any scalar arguments and if the
/// arguments can (ever) be null
///
/// See [`HigherOrderUDF::return_field_from_args`] for more information
#[derive(Clone, Debug)]
pub struct LambdaReturnFieldArgs<'a> {
    /// The data types of the arguments to the function
    ///
    /// If argument `i` to the function is a lambda, it will be the field of the result of the
    /// lambda if evaluated with the parameters returned from [`HigherOrderUDF::lambda_parameters`]
    ///
    /// For example, with `array_transform([1], v -> v == 5)`
    /// this field will be `[
    ///     ValueOrLambda::Value(Field::new("", DataType::List(DataType::Int32), false)),
    ///     ValueOrLambda::Lambda(Field::new("", DataType::Boolean, false))
    /// ]`
    pub arg_fields: &'a [ValueOrLambda<FieldRef, FieldRef>],
    /// Is argument `i` to the function a scalar (constant)?
    ///
    /// If the argument `i` is not a scalar, it will be None
    ///
    /// For example, if a function is called like `array_transform([1], v -> v == 5)`
    /// this field will be `[Some(ScalarValue::List(...), None]`
    pub scalar_arguments: &'a [Option<&'a ScalarValue>],
}

/// Arguments passed to [`HigherOrderUDF::invoke_with_args`] when invoking a
/// lambda function.
#[derive(Debug, Clone)]
pub struct HigherOrderFunctionArgs {
    /// The evaluated arguments and lambdas to the function
    pub args: Vec<ValueOrLambda<ColumnarValue, LambdaArgument>>,
    /// Field associated with each arg, if it exists
    /// For lambdas, it will be the field of the result of
    /// the lambda if evaluated with the parameters
    /// returned from [`HigherOrderUDF::lambda_parameters`]
    pub arg_fields: Vec<ValueOrLambda<FieldRef, FieldRef>>,
    /// The number of rows in record batch being evaluated
    pub number_rows: usize,
    /// The return field of the lambda function returned
    /// (from `return_field_from_args`) when creating the
    /// physical expression from the logical expression
    pub return_field: FieldRef,
    /// The config options at execution time
    pub config_options: Arc<ConfigOptions>,
}

/// A lambda argument to a HigherOrderFunction
#[derive(Clone, Debug)]
pub struct LambdaArgument {
    /// The parameters defined in this lambda
    ///
    /// For example, for `array_transform([2], v -> -v)`,
    /// this will be `vec![Field::new("v", DataType::Int32, true)]`
    params: Vec<FieldRef>,
    /// The body of the lambda
    ///
    /// For example, for `array_transform([2], v -> -v)`,
    /// this will be the physical expression of `-v`
    body: Arc<dyn PhysicalExpr>,
}

impl LambdaArgument {
    /// Evaluate this lambda
    /// `args` should evalute to the value of each parameter
    /// of the correspondent lambda returned in [HigherOrderUDF::lambda_parameters].
    pub fn evaluate(
        &self,
        args: &[&dyn Fn() -> Result<ArrayRef>],
    ) -> Result<ColumnarValue> {
        let columns = args
            .iter()
            .take(self.params.len())
            .map(|arg| arg())
            .collect::<Result<_>>()?;

        let schema = Arc::new(Schema::new(self.params.clone()));

        let batch = RecordBatch::try_new(schema, columns)?;

        self.body.evaluate(&batch)
    }
}
array_transform lambda_parameters implementation
impl HigherOrderUDF for ArrayTransform {
fn lambda_parameters(&self, value_fields: &[FieldRef]) -> Result<Vec<Vec<Field>>> {
        let list = if value_fields.len() == 1 {
            &value_fields[0]
        } else {
            return plan_err!(
                "{} function requires 1 value arguments, got {}",
                self.name(),
                value_fields.len()
            );
        };

        let field = match list.data_type() {
            DataType::List(field) => field,
            DataType::LargeList(field) => field,
            DataType::FixedSizeList(field, _) => field,
            _ => return plan_err!("expected list, got {list}"),
        };

        // we don't need to check whether the lambda contains more than two parameters,
        // e.g. array_transform([], (v, i, j) -> v+i+j), as datafusion will do that for us
        let value = Field::new("", field.data_type().clone(), field.is_nullable())
            .with_metadata(field.metadata().clone());

        Ok(vec![vec![value]])
    }
}
array_transform return_field_from_args implementation
fn value_lambda_pair<'a, V: Debug, L: Debug>(
    name: &str,
    args: &'a [ValueOrLambda<V, L>],
) -> Result<(&'a V, &'a L)> {
    let [value, lambda] = take_function_args(name, args)?;

    let (ValueOrLambda::Value(value), ValueOrLambda::Lambda(lambda)) = (value, lambda)
    else {
        return plan_err!(
            "{name} expects a value followed by a lambda, got {value:?} and {lambda:?}"
        );
    };

    Ok((value, lambda))
}

impl HigherOrderUDF for ArrayTransform {
    fn return_field_from_args(
        &self,
        args: HigherOrderReturnFieldArgs,
    ) -> Result<Arc<Field>> {
        let (list, lambda) = value_lambda_pair(self.name(), args.arg_fields)?;

        // lambda is the resulting field of executing the lambda body
        // with the parameters returned in lambda_parameters
        let field = Arc::new(Field::new(
            Field::LIST_FIELD_DEFAULT_NAME,
            lambda.data_type().clone(),
            lambda.is_nullable(),
        ));

        let return_type = match list.data_type() {
            DataType::List(_) => DataType::List(field),
            DataType::LargeList(_) => DataType::LargeList(field),
            DataType::FixedSizeList(_, size) => DataType::FixedSizeList(field, *size),
            other => plan_err!("expected list, got {other}")?,
        };

        Ok(Arc::new(Field::new("", return_type, list.is_nullable())))
    }
}
array_transform invoke_with_args implementation
impl HigherOrderUDF for ArrayTransform {
fn invoke_with_args(&self, args: HigherOrderFunctionArgs) -> Result<ColumnarValue> {
        let (list, lambda) = value_lambda_pair(self.name(), &args.args)?;

        let list_array = list.to_array(args.number_rows)?;

        // Fast path for fully null input array and also the only way to safely work with
        // a fully null fixed size list array as it can't be handled by remove_list_null_values below
        if list_array.null_count() == list_array.len() {
            return Ok(ColumnarValue::Array(new_null_array(
                args.return_type(),
                list_array.len(),
            )));
        }

        // as per list_values docs, if list_array is sliced, list_values will be sliced too,
        // so before constructing the transformed array below, we must adjust the list offsets with
        // adjust_offsets_for_slice
        let list_values = list_values(&list_array)?;

        // by passing closures, lambda.evaluate can evaluate only those actually needed
        let values_param = || Ok(Arc::clone(&list_values));

        // call the transforming lambda
        let transformed_values = lambda
            .evaluate(&[&values_param])?
            .into_array(list_values.len())?;

        let field = match args.return_field.data_type() {
            DataType::List(field)
            | DataType::LargeList(field)
            | DataType::FixedSizeList(field, _) => Arc::clone(field),
            _ => {
                return exec_err!(
                    "{} expected ScalarFunctionArgs.return_field to be a list, got {}",
                    self.name(),
                    args.return_field
                );
            }
        };

        let transformed_list = match list_array.data_type() {
            DataType::List(_) => {
                let list = list_array.as_list();

                // since we called list_values above which would return sliced values for
                // a sliced list, we must adjust the offsets here as otherwise they would be invalid
                let adjusted_offsets = adjust_offsets_for_slice(list);

                Arc::new(ListArray::new(
                    field,
                    adjusted_offsets,
                    transformed_values,
                    list.nulls().cloned(),
                )) as ArrayRef
            }
            DataType::LargeList(_) => {
                let large_list = list_array.as_list();

                // since we called list_values above which would return sliced values for
                // a sliced list, we must adjust the offsets here as otherwise they would be invalid
                let adjusted_offsets = adjust_offsets_for_slice(large_list);

                Arc::new(LargeListArray::new(
                    field,
                    adjusted_offsets,
                    transformed_values,
                    large_list.nulls().cloned(),
                ))
            }
            DataType::FixedSizeList(_, value_length) => {
                Arc::new(FixedSizeListArray::new(
                    field,
                    *value_length,
                    transformed_values,
                    list_array.as_fixed_size_list().nulls().cloned(),
                ))
            }
            other => exec_err!("expected list, got {other}")?,
        };

        Ok(ColumnarValue::Array(transformed_list))
    }
}
How relevant HigherOrderUDF methods would be called and what they would return during planning and evaluation of the example
// this is called at sql planning
let lambda_parameters = lambda_udf.lambda_parameters(&[
    Field::new("", DataType::new_list(DataType::Int32, false), false), // the Field of the [2, 3] literal
])?;

assert_eq!(
    lambda_parameters,
    vec![
            // the parameters that *can* be declared on the lambda, and not only 
            // those actually declared: the implementation doesn't need to care 
            // about it
            vec![
                Field::new("", DataType::Int32, false), // the list inner value
            ]]
);



// this is called every time ExprSchemable is called on a HigherOrderFunction
let return_field = array_transform.return_field_from_args(&LambdaReturnFieldArgs {
    arg_fields: &[
        ValueOrLambda::Value(Field::new("", DataType::new_list(DataType::Int32, false), false)),
        ValueOrLambda::Lambda(Field::new("", DataType::Boolean, false)), // the return_field of the expression "v != 2" when "v" is of the type returned in lambda_parameters
    ],
    scalar_arguments // irrelevant
})?;

assert_eq!(return_field, Field::new("", DataType::new_list(DataType::Boolean, false), false));



let value = array_transform.evaluate(&HigherOrderFunctionArgs {
    args: vec![
        ValueOrLambda::Value(List([2, 3])),
        ValueOrLambda::Lambda(LambdaArgument of `v -> v != 2`),
    ],
    arg_fields, // same as above
    number_rows: 1,
    return_field, // same as above
    config_options, // irrelevant
})?;

assert_eq!(value, BooleanArray::from([false, true]))


A pair HigherOrderUDF/HigherOrderUDFImpl like ScalarFunction was not used because those exist only to maintain backwards compatibility with the older API #8045


Why LambdaVariable and not Column:

Existing tree traversals that operate on columns would break if some column nodes referenced to a lambda parameter and not a real column. In the example query, projection pushdown would try to push the lambda parameter "v", which won't exist in table "t".

Example of code of another traversal that would break:

fn minimize_join_filter(expr: Arc<dyn PhysicalExpr>, ...) -> JoinFilter {
    let mut used_columns = HashSet::new();
    expr.apply(|expr| {
        if let Some(col) = expr.as_any().downcast_ref::<Column>() {
            // if this is a lambda column, this function will break
            used_columns.insert(col.index());
        }
        Ok(TreeNodeRecursion::Continue)
    });
    ...
}

Furthermore, the implemention of ExprSchemable and PhysicalExpr::return_field for Column expects that the schema it receives as a argument contains an entry for its name, which is not the case for lambda parameters.

By including a FieldRef on LambdaVariable that should be resolved during construction time in the sql planner, ExprSchemable and PhysicalExpr::return_field simply return it's own Field:

LambdaVariable ExprSchemable and PhysicalExpr::return_field implementation
impl ExprSchemable for Expr {
   fn to_field(
        &self,
        schema: &dyn ExprSchema,
    ) -> Result<(Option<TableReference>, Arc<Field>)> {
        let (relation, schema_name) = self.qualified_name();
        let field = match self {
           Expr::LambdaVariable(l) => Ok(Arc::clone(&l.field)),
           ...
        }?;

        Ok((
            relation,
            Arc::new(field.as_ref().clone().with_name(schema_name)),
        ))
    }
    ...
}

impl PhysicalExpr for LambdaVariable {
    fn return_field(&self, _input_schema: &Schema) -> Result<FieldRef> {
        Ok(Arc::clone(&self.field))
    }
    ...
}

Possible alternatives discarded due to complexity, requiring downstream changes and implementation size:
  1. Add a new set of TreeNode methods that provides the set of lambdas parameters names seen during the traversal, so column nodes can be tested if they refer to a regular column or to a lambda parameter. Any downstream user that wants to support lambdas would need use those methods instead of the existing ones. This also would add 1k+ lines to the PR.
impl Expr {
    pub fn transform_with_lambdas_params<
        F: FnMut(Self, &HashSet<String>) -> Result<Transformed<Self>>,
    >(
        self,
        mut f: F,
    ) -> Result<Transformed<Self>> {}
}

How minimize_join_filter would looks like:

fn minimize_join_filter(expr: Arc<dyn PhysicalExpr>, ...) -> JoinFilter {
    let mut used_columns = HashSet::new();
    expr.apply_with_lambdas_params(|expr, lambdas_params| {
        if let Some(col) = expr.as_any().downcast_ref::<Column>() {
            // dont include lambdas parameters
            if !lambdas_params.contains(col.name()) {
                used_columns.insert(col.index());
            }
        }
        Ok(TreeNodeRecursion::Continue)
    })
    ...
}
  1. Add a flag to the Column node indicating if it refers to a lambda parameter. Still requires checking for it on existing tree traversals that works on Columns (30+) and also downstream.
//logical
struct Column {
    pub relation: Option<TableReference>,
    pub name: String,
    pub spans: Spans,
    pub is_lambda_parameter: bool,
}

//physical
struct Column {
    name: String,
    index: usize,
    is_lambda_parameter: bool,
}

How minimize_join_filter would look like:

fn minimize_join_filter(expr: Arc<dyn PhysicalExpr>, ...) -> JoinFilter {
    let mut used_columns = HashSet::new();
    expr.apply(|expr| {
        if let Some(col) = expr.as_any().downcast_ref::<Column>() {
            // dont include lambdas parameters
            if !col.is_lambda_parameter {
                used_columns.insert(col.index());
            }
        }
        Ok(TreeNodeRecursion::Continue)
    })
    ...
}
  1. Add a new set of TreeNode methods that provides a schema that includes the lambdas parameters for the scope of the node being visited/transformed:
impl Expr {
    pub fn transform_with_schema<
        F: FnMut(Self, &DFSchema) -> Result<Transformed<Self>>,
    >(
        self,
        schema: &DFSchema,
        f: F,
    ) -> Result<Transformed<Self>> { ... }
    ... other methods
}

For any given HigherOrderFunction found during the traversal, a new schema is created for each lambda argument that contains it's parameter, returned from HigherOrderUDF::lambda_parameters
How it would look like:

pub fn infer_placeholder_types(self, schema: &DFSchema) -> Result<(Expr, bool)> {
        let mut has_placeholder = false;
        // Provide the schema as the first argument. 
        // Transforming closure receive an adjusted_schema as argument
        self.transform_with_schema(schema, |mut expr, adjusted_schema| {
            match &mut expr {
                // Default to assuming the arguments are the same type
                Expr::BinaryExpr(BinaryExpr { left, op: _, right }) => {
                    // use adjusted_schema and not schema. Those expressions may contain 
                    // columns referring to a lambda parameter, which Field would only be
                    // available in adjusted_schema and not in schema
                    rewrite_placeholder(left.as_mut(), right.as_ref(), adjusted_schema)?;
                    rewrite_placeholder(right.as_mut(), left.as_ref(), adjusted_schema)?;
                }
    ....
  1. Make available trought LogicalPlan and ExecutionPlan nodes a schema that includes all lambdas parameters from all expressions owned by the node, and use this schema for tree traversals. For nodes which won't own any expression, the regular schema can be returned
impl LogicalPlan {
    fn lambda_extended_schema(&self) -> &DFSchema;
}

trait ExecutionPlan {
    fn lambda_extended_schema(&self) -> &DFSchema;
}

//usage
impl LogicalPlan {
    pub fn replace_params_with_values(
            self,
            param_values: &ParamValues,
        ) -> Result<LogicalPlan> {
            self.transform_up_with_subqueries(|plan| {
                // use plan.lambda_extended_schema() containing lambdas parameters
                // instead of plan.schema() which wont
                let lambda_extended_schema = Arc::clone(plan.lambda_extended_schema());
                let name_preserver = NamePreserver::new(&plan);
                plan.map_expressions(|e| {
                    // if this expression is child of lambda and contain columns referring it's parameters
                    // the lambda_extended_schema already contain them
                    let (e, has_placeholder) = e.infer_placeholder_types(&lambda_extended_schema)?;
    ....

Comment on lines +211 to +215
/// A RecordBatch with the captured columns inside the lambda body, if any
///
/// For example, for `array_transform([2], v -> v + a + b)`,
/// this will be a `RecordBatch` with columns `a` and `b`
captures: Option<RecordBatch>,
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dont think that we should have this for column capture and I think it would be a long review for the design on the column capture itself, so I'm fine for not including this preparation for column capture now and create a breaking change in a different pr (I'm also referring to the rest of the breaking changes made to support column capture like the evaluate added argument)

for example I think we should have ProjectionExpr, and use the existing logic for projecting, and not get all the arrays in the evaluate function

Copy link
Copy Markdown
Contributor Author

@gstvg gstvg Apr 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree, reverted at 9cb4882
cc @LiaCastaneda

pub fn evaluate(
&self,
args: &[&dyn Fn() -> Result<ArrayRef>],
_adjust: impl FnOnce(&[ArrayRef]) -> Result<Vec<ArrayRef>>,
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would revert this as well as explained before

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment on lines +305 to +309
/// The return of [HigherOrderUDF::lambda_parameters]
pub enum LambdaParametersProgress {
Partial(Vec<Option<Vec<FieldRef>>>),
Complete(Vec<Vec<FieldRef>>),
}
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this needs more comments explaining what each variant is, and when you should use each one (I know that you added some comments regarding this on lambda_parameters but some should also be here)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Indeed, is much better now, thanks 15a0106

Lambda(L),
}

/// The return of [HigherOrderUDF::lambda_parameters]
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the comment should explain the enum and not the function that return it IMO

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed, also at 15a0106

@rluvaton rluvaton dismissed their stale review April 27, 2026 08:54

have new important comments

Copy link
Copy Markdown
Contributor

@LiaCastaneda LiaCastaneda left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree we should move the column capture support to a follow up PR, since it would involve more design discussions and this PR already proves that most lambda functions can be implemented easily -- I don't think we should block on it. Left some comments on the multi step type resolution you introduced in the last commits

Comment thread datafusion/sql/src/expr/function.rs Outdated
Comment on lines +518 to +524
if step > 256 {
return plan_err!(
"{} lambda_parameters called 256 times without completion",
fm.name()
);
}
};
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤔 This limit feels arbitrary, even if I can't imagine a situation where a user has 256+ lambdas in a higher-order function each outputting a different type, in theory implementors express this. Since implementors know beforehand how their higher-order function typing works, it might make sense to include something like max_lambda_parameter_steps: usize in HigherOrderSignature, defaulting to 1?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It misses a comment, indeed. It's mostly a guard against a faulty lambda_parameters that causes an infinite loop by endlessly returnin Partial's, but I agree that making it configurable it's better, but I stick to 256 as default so that ideally nobody has ever to think about it 9f19fed

Comment on lines +545 to +548
fn coerce_values_for_lambdas(
&self,
_fields: &[ValueOrLambda<DataType, DataType>],
) -> Result<Vec<DataType>> {
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

coerce_values_for_lambdas and coerce_value_types do the same thing, but one is called after the lambdas are planned right? iiuc, you added a separate function because you need to see both lambdas and values via ValueOrLambda. If that's the only reason for having a separate function, maybe we should just have one function that receives &[ValueOrLambda<DataType, DataType>]? In the other PR I suggested having only &[DataType] because I couldn't think of a near term use case that would actually need both.

Not sure what the cleaner approach is here -- having two APIs that do the same thing, or having one API with &[ValueOrLambda<DataType, DataType>] as the parameter.

also I see this function is only called if coerce_values_for_lambdas flag is set to true in the signature, which can be easy to forget to the implementor.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is also missing an explanation. While unlikely, is still valid that coerce_value_types coercion changes the input values in a way that also changes lambda parameters: most cases will coerce from ListView(Int32) to List(Int32), for example (the lambda parameter is still Int32), but it also may coerce into a List(Int64) (the lambda parameter is Int64 now). So we need to call coerce_value_types before calling lambda_parameters, and then use it's output to call coerce_values_for_lambdas. 8f3b0fe

Copy link
Copy Markdown
Contributor Author

@gstvg gstvg Apr 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the other PR I suggested having only &[DataType] because I couldn't think of a near term use case that would actually need both.

I believe at that time it was ValueOrLambda<DataType, ()> only to detect bad arg positions but that function is likely not the place for it , so your suggestion still makes sense even taking into account the array_reduce signature.

also I see this function is only called if coerce_values_for_lambdas flag is set to true in the signature, which can be easy to forget to the implementor

Yeah, thats true. When adding it I thought about removing the flag, changing the return type to Result<Option<Vec<DataType>>> with a blank impl Ok(None), so that it would always be called but do nothing with the default impl, but I didn't like much the signature, WDYT?

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is also missing an explanation. While unlikely, is still valid that coerce_value_types coercion changes the input values in a way that also changes lambda parameters: most cases will coerce from ListView(Int32) to List(Int32), for example (the lambda parameter is still Int32), but it also may coerce into a List(Int64) (the lambda parameter is Int64 now). So we need to call coerce_value_types before calling lambda_parameters, and then use it's output to call coerce_values_for_lambdas. 8f3b0fe

is there a test for that?

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@LiaCastaneda after this is resolved and no further comments I'll merge this PR

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

expanded an existing test to cover this c93fc81

Copy link
Copy Markdown
Contributor

@LiaCastaneda LiaCastaneda Apr 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, thats true. When adding it I thought about removing the flag, changing the return type to Result<Option<Vec>> with a blank impl Ok(None), so that it would always be called but do nothing with the default impl, but I didn't like much the signature, WDYT?

I understand. imo even if not the prettiest signature, Result<Option<Vec<DataType>>> removes an extra thing the implementor has to remember, the contract is in the return type itself. There are other traits/apis in DataFusion that also have a return of type Result<Option<...>>, so I guess it's fine to have it like that

@comphead
Copy link
Copy Markdown
Contributor

I tried to build array_exists spark function using current approach #21881

Good thing it works. From the performance point of view there are some improvements can be made for lambdas but likely we can address it in follow, @gstvg it is up to you

 Loops — hoisting & fusion                                                                                                                                                                                
                                                                                                                                                                                                           
  L1. Three passes over self.args in one evaluate call. datafusion/physical-expr/src/higher_order_function.rs:247, :269, :309 each iterate self.args, and each of the first and third pass calls           
  self.lambda_positions.contains(&i) (linear scan) inside the loop body. For N args with L lambdas, that's O(N·L) per batch just to classify args — the classification is already known at construction    
  time.                                                     
                                                                                                                                                                                                           
  - Fix: Replace lambda_positions: Vec<usize> with a per-arg enum cached on the struct:                                                                                                                    
  enum ArgSlot { Value(Arc<dyn PhysicalExpr>), Lambda(Arc<LambdaExpr>) }
  slots: Vec<ArgSlot>,                                                                                                                                                                                     
  - The three passes then become one fused loop; O(N) per batch, zero allocation for the lookup, and the slow wrapped_lambda tree walk (:459-474) collapses to a cached Arc<LambdaExpr>.                   
                                                                                                                                                                                                           
  L2. Pass 2 is pure reshaping of pass 1 (arg_fields → fields, cloning Arcs). Can be folded into pass 1 by building the Vec<ValueOrLambda<FieldRef, Option<FieldRef>>> directly.                           
                                                                                                                                                                                                           
  ---                                                                                                                                                                                                      
  Data structures — precompute / augment                    
                                                                                                                                                                                                           
  D1. LambdaArgument::evaluate rebuilds the inner Schema on every call. datafusion/expr/src/higher_order_function.rs:256:
  let schema = Arc::new(Schema::new(self.params.clone()));                                                                                                                                                 
  Schema::new builds a name→index HashMap internally. For a nested HOF (array_transform([[1,2]], a -> array_transform(a, b -> ...))) this runs once per outer sublist per batch. Build it once in
  LambdaArgument::new and store Arc<Schema>.                                                                                                                                                               
                                                                                                                                                                                                           
  D2. self.fun.clear_null_values() is called once per non-lambda arg per batch (:349). It's a &self method returning a constant bool. Cache as a bool on HigherOrderFunctionExpr at construction.          
                                                                                                                                                                                                           
  D3. wrapped_lambda walks the expr tree every evaluate. :253 and :315 both call it. Pre-resolve the Arc<LambdaExpr> once at try_new_with_schema / with_new_children and store it — lambdas don't change   
  mid-execution.                                                                                                                                                                                           
                                                                                                                                                                                                           
  D4. HigherOrderFunctionArgs::args and ::arg_fields are always Vec even for 2-arg HOFs (array_transform, array_exists, array_filter — the vast majority). Switching to smallvec::SmallVec<[_; 2]>         
  eliminates a heap alloc per batch for the common shape.
                                                                                                                                                                                                           
  ---                                                       
  Logic — short-circuit / reorder / lazy
                                                                                                                                                                                                           
  G1. lambda_parameters(0, &fields) runs unconditionally at :285 even when there are no lambdas. Guard with if self.lambda_positions.is_empty() { skip }.
                                                                                                                                                                                                           
  G2. conditional_arguments is specced but unused by the evaluator. The trait exposes short_circuits() + conditional_arguments(args) to mark args as lazy, but HigherOrderFunctionExpr::evaluate still     
  eagerly arg.evaluate(batch) for every non-lambda value (:346). A short-circuiting array_exists/array_any can't benefit from skipping remaining work today. Wire this up.                                 
                                                                                                                                                                                                           
  G3. remove_list_null_values path (:349-355) is entered for every List/LargeList arg. The callee has a null_count == 0 fast path that returns list.clone(), but you still pay an outer Arc::new+match. Add
   array.null_count() == 0 short-circuit at the caller.
                                                                                                                                                                                                           
  G4. Constant-body lambdas aren't specialized. array_transform([..5], v -> 42) builds a RecordBatch and evaluates the body over all values despite the body not referencing v. Detect "no LambdaVariable  
  in body" at construction; evaluate the body on an empty batch once and replicate.
                                                                                                                                                                                                           
  ---                                                                                                                                                                                                      
  Expressions — word/SIMD parallelism & CSE
                                                                                                                                                                                                           
  E1. Per-row predicate scans are elementwise today (see my compute_exists in array_exists.rs and the array_transform pattern). Replace the for i in start..end { if predicate.value(i) … } loops with
  Arrow's bit-level operations on the predicate's underlying BooleanBuffer:                                                                                                                                
                                                            
  let row_slice = predicate.slice(start, end - start);                                                                                                                                                     
  if row_slice.true_count() > 0   { TRUE }            // SIMD popcount                                                                                                                                     
  else if row_slice.null_count() > 0 { NULL }                                                                                                                                                              
  else                               { FALSE }                                                                                                                                                             
                                                                                                                                                                                                           
  Same trick applies to any boolean-reducing HOF (exists, forall, filter size estimation).                                                                                                                 
                                                                                                                                                                                                           
  E2. self.lambda_positions.contains(&i) is a repeated CSE target (L1 above covers it).                                                                                                                    
                                                            
  ---                                                                                                                                                                                                      
  Procedures — parallelism                                  
                                                                                                                                                                                                           
  P1. Flat lambda evaluation over list_values is embarrassingly parallel. For per-row-independent HOFs (transform, exists, filter, any, forall), list_values can be chunked and the lambda body evaluated
  in parallel per chunk, then concatenated. Reuses Arrow's kernels that already release the GIL equivalent. Gates on detecting side-effect-free lambdas (e.g. !is_volatile_node()).                        
                                                            
  P2. No HOF benchmarks exist (rg "HigherOrder" benches/ returns nothing). Before any of the above, add a Criterion bench for array_transform over (List, LargeList, FixedSizeList × 10K/1M rows ×         
  scalar/columnar lambda body). Without baseline numbers, "faster" is unmeasurable.
                                                                                                                                                                                                           
  ---                                                       
  Suggested priority order
                                                                                                                                                                                                           
  1. D3 + L1 + L2 (cache ArgSlot, fold loops) — one edit, removes the O(N·L) classification and two Vec reallocs per batch.
  2. D1 (cache lambda Schema) — one-line fix, big win for nested HOFs.                                                                                                                                     
  3. E1 (SIMD popcount on predicate boolean buffer) — biggest per-element win on array_exists / filter / forall.                                                                                           
  4. P2 (add Criterion benches) — required to validate any of the above.                                                                                                                                   
  5. G2 (wire up conditional_arguments) — needed before short-circuiting HOFs like exists can skip work.

@gstvg
Copy link
Copy Markdown
Contributor Author

gstvg commented Apr 28, 2026

@comphead That's great. I'm almost sure none of these improvements are breaking so we can address it on another PR. I added it to #21172

CI failure seems unrelated: https://github.com/apache/datafusion/actions/runs/25037217756/job/73331583409#step:7:919

@rluvaton
Copy link
Copy Markdown
Member

@comphead Regarding the performance optimization, I'm adding some helpers in arrow-rs that will handle sliced and cleanup of nulls to make it super fast than we can use later

Copy link
Copy Markdown
Member

@rluvaton rluvaton left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well done

@rluvaton
Copy link
Copy Markdown
Member

@comphead looks like you have no other comments, and I approved, so I will wait for the last @LiaCastaneda comment to be resolved and I'll merge this

@rluvaton rluvaton added this pull request to the merge queue Apr 28, 2026
Merged via the queue into apache:main with commit bbf67d9 Apr 28, 2026
40 checks passed
@rluvaton
Copy link
Copy Markdown
Member

Big thank you for @gstvg and everyone involved!

@comphead
Copy link
Copy Markdown
Contributor

Thanks everyone! This is a huge one!

@alamb
Copy link
Copy Markdown
Contributor

alamb commented Apr 28, 2026

this is pretty amazing -- I put a note to include it in the 55 release's notes:

github-merge-queue Bot pushed a commit that referenced this pull request Apr 28, 2026
## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #.

Followup on
#21679 (comment)

## Rationale for this change

<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->

## What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->

## Are these changes tested?

<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

## Are there any user-facing changes?

<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->

<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->

---------

Co-authored-by: Raz Luvaton <16746759+rluvaton@users.noreply.github.com>
@gstvg
Copy link
Copy Markdown
Contributor Author

gstvg commented Apr 29, 2026

Many thanks for everyone involved, both reviewers and also those who showed interest on the feature. Reviewing such a big PR is not easy and I'm very grateful for it, thanks again ❤️

@timsaucer
Copy link
Copy Markdown
Member

This is amazing. Thank you so much for all the hard work on this. This is definitely one of the things I am most excited about in the next release. This is going to be huge!

@adriangb
Copy link
Copy Markdown
Contributor

adriangb commented May 1, 2026

I just wanted to bring to attention that DuckDB deprecated this very sytnax because of conflicts with JSON operators.

I guess we may want to support both long term (Spark uses the arrow syntax), but I think there's a real risk that we are not even evaluating incompatibility with JSON operators because they are not implemented by default in DataFusion (but we are in talks to do so #21301).

Happy to open an issue for discussion but wanted to check first if this was discussed at all, as far as I can tell from going over the PR it has not.

@gstvg
Copy link
Copy Markdown
Contributor Author

gstvg commented May 1, 2026

@adriangb I think this decision can be left to the user via the configurable dialect, as today. This PR merely consumes the LambdaFunction from sqlparser-rs AST, which syntax it parses is defined by the configured dialect. Is up to sqlparser to avoid conflicts (see apache/datafusion-sqlparser-rs#2224). The sqllogictests here requires setting the dialect to databricks, for example.

I guess what we can do here is:

  • Add examples to the existing higher-order functions docs with the duckdb syntax with a note that it's dialect dependent. And do the same for json operators as well when they land
  • Format lambdas with the duckdb syntax, since it's unambiguous and because we don't have access to the configured dialect during formatting
  • Fix the sql unparser to return a dialect dependent syntax (it currently returns the old syntax for duckdb)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

api change Changes the API exposed to users of the crate catalog Related to the catalog crate common Related to common crate core Core DataFusion crate datasource Changes to the datasource crate documentation Improvements or additions to documentation execution Related to the execution crate ffi Changes to the ffi crate functions Changes to functions implementation logical-expr Logical plan and expressions optimizer Optimizer rules physical-expr Changes to the physical-expr crates proto Related to proto crate spark sql SQL Planner sqllogictest SQL Logic Tests (.slt) substrait Changes to the substrait crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

10 participants